package archive

import (
	"context"
	"fmt"
	"log"
	"strings"
	"time"

	"gitlab.com/gitlab-org/gitaly/v14/internal/gitaly/archive"
	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// ManageRefs implements the ManageRefs RPC method
func (s *server) ManageRefs(ctx context.Context, req *gitalypb.ManageRefsRequest) (*gitalypb.ManageRefsResponse, error) {
	if req.GetRepository() == nil {
		return nil, status.Error(codes.InvalidArgument, "repository is required")
	}

	if len(req.GetRefs()) == 0 {
		return nil, status.Error(codes.InvalidArgument, "refs is required")
	}

	// Validate action
	switch req.GetAction() {
	case gitalypb.ArchiveOperation_ARCHIVE, gitalypb.ArchiveOperation_RESTORE, gitalypb.ArchiveOperation_DELETE:
		// Valid actions
	default:
		return nil, status.Error(codes.InvalidArgument, "action is required and must be ARCHIVE, RESTORE, or DELETE")
	}

	// Validate input based on action
	if err := s.validateRefsForAction(req); err != nil {
		return nil, err
	}

	// Get repository path
	repoPath, err := s.locator.GetRepoPath(req.GetRepository())
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get repository path: %v", err)
	}

	// Convert request to metadata entries
	now := time.Now()
	var entries []archive.MetadataEntry
	var failedRefs []string

	// If this is a restore operation with delete_archive_refs=true, create two sets of entries
	if req.GetAction() == gitalypb.ArchiveOperation_RESTORE && req.GetDeleteArchiveRefs() {
		// Create restore entries
		for _, ref := range req.GetRefs() {
			entry, err := s.createMetadataEntry(req, ref, now)
			if err != nil {
				failedRefs = append(failedRefs, ref.GetSourceRefName())
				continue
			}
			entries = append(entries, entry)
		}

		// Create delete entries (for the same refs but as delete operations)
		deleteReq := &gitalypb.ManageRefsRequest{
			Repository:  req.GetRepository(),
			Action:      gitalypb.ArchiveOperation_DELETE,
			Refs:        req.GetRefs(),
			CreatorId:   req.GetCreatorId(),
			CreatorName: req.GetCreatorName(),
			Reason:      req.GetReason() + " (auto-delete after restore)",
		}

		for _, ref := range req.GetRefs() {
			deleteEntry, err := s.createMetadataEntry(deleteReq, ref, now)
			if err != nil {
				failedRefs = append(failedRefs, ref.GetSourceRefName())
				continue
			}
			entries = append(entries, deleteEntry)
		}
	} else {
		// Normal processing for single operation
		for _, ref := range req.GetRefs() {
			entry, err := s.createMetadataEntry(req, ref, now)
			if err != nil {
				failedRefs = append(failedRefs, ref.GetSourceRefName())
				continue
			}
			entries = append(entries, entry)
		}
	}

	// Generate unique task IDs for this batch
	restoreTaskID := generateTaskID(gitalypb.ArchiveOperation_RESTORE, req.GetCreatorId(), now)
	var allTaskIDs []string

	if req.GetAction() == gitalypb.ArchiveOperation_RESTORE && req.GetDeleteArchiveRefs() {
		// Split entries into restore and delete groups
		var restoreEntries []archive.MetadataEntry
		var deleteEntries []archive.MetadataEntry

		for _, entry := range entries {
			if entry.Action == archive.ActionRestore {
				restoreEntries = append(restoreEntries, entry)
			} else if entry.Action == archive.ActionDelete {
				deleteEntries = append(deleteEntries, entry)
			}
		}

		// Write restore metadata
		if len(restoreEntries) > 0 {
			if err := s.storageManager.WriteMetadata(repoPath, restoreEntries, archive.StatusWaiting, restoreTaskID); err != nil {
				return nil, status.Errorf(codes.Internal, "failed to write restore metadata: %v", err)
			}
			allTaskIDs = append(allTaskIDs, restoreTaskID)
		}

		// Write delete metadata with different task ID
		if len(deleteEntries) > 0 {
			deleteTaskID := generateTaskID(gitalypb.ArchiveOperation_DELETE, req.GetCreatorId(), now)
			if err := s.storageManager.WriteMetadata(repoPath, deleteEntries, archive.StatusWaiting, deleteTaskID); err != nil {
				return nil, status.Errorf(codes.Internal, "failed to write delete metadata: %v", err)
			}
			allTaskIDs = append(allTaskIDs, deleteTaskID)
		}
	} else {
		// Normal single operation
		mainTaskID := generateTaskID(req.GetAction(), req.GetCreatorId(), now)
		if err := s.storageManager.WriteMetadata(repoPath, entries, archive.StatusWaiting, mainTaskID); err != nil {
			return nil, status.Errorf(codes.Internal, "failed to write metadata: %v", err)
		}
		allTaskIDs = append(allTaskIDs, mainTaskID)
	}

	// Send tasks to Kafka for processing
	if s.kafkaProducer != nil && s.kafkaProducer.IsEnabled() {
		for _, taskID := range allTaskIDs {
			// Determine action for this task
			var action archive.OperationAction
			if strings.Contains(taskID, "RESTORE") {
				action = archive.ActionRestore
			} else if strings.Contains(taskID, "DELETE") {
				action = archive.ActionDelete
			} else if strings.Contains(taskID, "ARCHIVE") {
				action = archive.ActionArchive
			}

			// 获取引用类型
			var refType string
			if len(entries) > 0 {
				refType = archive.ParseRefType(entries[0].SourceRefName)
			}

			// 构建元数据文件名
			metadataFileName := fmt.Sprintf("refs-list-%d-%s-%s.json", now.Unix(), taskID, action)

			if err := s.kafkaProducer.SendTaskMessage(ctx, taskID, repoPath, req.GetRepository().GetStorageName(), metadataFileName); err != nil {
				// Log error but don't fail the request - the task can be processed later
				log.Printf("Failed to send task %s to Kafka: %v", taskID, err)
			} else {
				log.Printf("Successfully sent task %s to Kafka for action: %s", taskID, action)
			}
		}
	}

	return &gitalypb.ManageRefsResponse{
		ProcessedCount: int32(len(req.GetRefs())), // Count original refs, not duplicate entries
		FailedRefs:     failedRefs,
		TaskId:         strings.Join(allTaskIDs, ","), // Return all task IDs
	}, nil
}

// validateRefsForAction validates refs based on the action
func (s *server) validateRefsForAction(req *gitalypb.ManageRefsRequest) error {
	switch req.GetAction() {
	case gitalypb.ArchiveOperation_ARCHIVE:
		// Archive operation: source_ref_name and source_commit_id are required
		for i, ref := range req.GetRefs() {
			if ref.GetSourceRefName() == "" {
				return status.Errorf(codes.InvalidArgument, "refs[%d].source_ref_name is required for archive operation", i)
			}
			if ref.GetSourceCommitId() == "" {
				return status.Errorf(codes.InvalidArgument, "refs[%d].source_commit_id is required for archive operation", i)
			}
		}
	case gitalypb.ArchiveOperation_RESTORE:
		// Restore operation: source_ref_name, source_commit_id and target_ref_name are required
		for i, ref := range req.GetRefs() {
			if ref.GetSourceRefName() == "" {
				return status.Errorf(codes.InvalidArgument, "refs[%d].source_ref_name is required for restore operation", i)
			}
			if ref.GetSourceCommitId() == "" {
				return status.Errorf(codes.InvalidArgument, "refs[%d].source_commit_id is required for restore operation", i)
			}
			if ref.GetTargetRefName() == "" {
				return status.Errorf(codes.InvalidArgument, "refs[%d].target_ref_name is required for restore operation", i)
			}
		}
	case gitalypb.ArchiveOperation_DELETE:
		// Delete operation: source_ref_name and source_commit_id are required
		for i, ref := range req.GetRefs() {
			if ref.GetSourceRefName() == "" {
				return status.Errorf(codes.InvalidArgument, "refs[%d].source_ref_name is required for delete operation", i)
			}
			if ref.GetSourceCommitId() == "" {
				return status.Errorf(codes.InvalidArgument, "refs[%d].source_commit_id is required for delete operation", i)
			}
		}
	}
	return nil
}

// createMetadataEntry creates a metadata entry from request data
func (s *server) createMetadataEntry(req *gitalypb.ManageRefsRequest, ref *gitalypb.ArchiveRef, now time.Time) (archive.MetadataEntry, error) {
	var action archive.OperationAction
	switch req.GetAction() {
	case gitalypb.ArchiveOperation_ARCHIVE:
		action = archive.ActionArchive
	case gitalypb.ArchiveOperation_RESTORE:
		action = archive.ActionRestore
	case gitalypb.ArchiveOperation_DELETE:
		action = archive.ActionDelete
	default:
		return archive.MetadataEntry{}, fmt.Errorf("unknown action: %v", req.GetAction())
	}

	entry := archive.MetadataEntry{
		Action:         action,
		SourceRefName:  ref.GetSourceRefName(),
		SourceCommitID: ref.GetSourceCommitId(),
		TargetRefName:  ref.GetTargetRefName(),
		CreatorID:      req.GetCreatorId(),
		CreatorName:    req.GetCreatorName(),
		CreateAt:       now,
		UpdateAt:       now,
		Status:         archive.StatusWaiting,
		Reason:         req.GetReason(),
	}

	// For archive operations, generate target ref name if not provided
	if action == archive.ActionArchive && entry.TargetRefName == "" {
		entry.TargetRefName = archive.GenerateArchivedRefName(ref.GetSourceRefName(), ref.GetSourceCommitId())
	}

	return entry, nil
}

// generateTaskID generates a unique task ID
func generateTaskID(action gitalypb.ArchiveOperation_Action, creatorID int64, timestamp time.Time) string {
	actionStr := action.String()
	return fmt.Sprintf("%s-%d-%d", actionStr, creatorID, timestamp.Unix())
}
